package com.zh.rocketmq;

import org.apache.rocketmq.client.apis.ClientConfiguration;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.ClientServiceProvider;
import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
import org.apache.rocketmq.client.apis.consumer.FilterExpression;
import org.apache.rocketmq.client.apis.consumer.FilterExpressionType;
import org.apache.rocketmq.client.apis.consumer.PushConsumer;

import java.io.IOException;
import java.util.Collections;

/**
 * @author Zhou
 * @date 16:25 2025/10/16
 */
public class ConsumerExample {
    public static void main(String[] args) throws ClientException, IOException, InterruptedException {
        // 接入点地址，需要设置成Proxy的地址和端口列表，一般是xxx:8081;xxx:8081。
        var endpoint = "172.24.103.123:8081";

        ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder()
                .setEndpoints(endpoint)
                .build();

        final ClientServiceProvider provider = ClientServiceProvider.loadService();

        // 为消费者指定所属的消费者分组，Group需要提前创建。
        var consumerGroup = "ConsumerOneGroup";
        // 指定需要订阅哪个目标Topic，Topic需要提前创建。
        var topic = "DemoTopic";
        // 订阅消息的过滤规则，"*"表示订阅所有Tag的消息。
        var filterExpression = new FilterExpression("*", FilterExpressionType.TAG);

        // 初始化PushConsumer，需要绑定消费者分组ConsumerGroup、通信参数以及订阅关系。
        PushConsumer pushConsumer = provider.newPushConsumerBuilder()
                .setClientConfiguration(clientConfiguration)
                .setConsumerGroup(consumerGroup)
                .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression)) // 设置预绑定的订阅关系。
                // 设置消费监听器。
                .setMessageListener(messageView -> {
                    // 处理消息并返回消费结果。
                    System.out.println("Consume message successfully, messageId= "+ messageView.getMessageId());
                    return ConsumeResult.SUCCESS;
                }).build();

        Thread.sleep(Integer.MAX_VALUE);

        // 如果不需要再使用 PushConsumer，可关闭该实例。
         pushConsumer.close();
    }
}
